const { Readable, Writable } = require('node:stream');
const fs = require('node:fs');
const path = require('path');

// 检查文件读权限
function checkReadPermission(filePath, callback) {
  fs.access(filePath, fs.constants.R_OK, (err) => {
      console.log('--------------', err)
    if (err) {
      callback(err);
    } else {
      callback(null);
    }
  });
}

// 检查目录写权限
function checkWritePermission(dirPath, callback) {
  fs.access(dirPath, fs.constants.W_OK, (err) => {
    if (err) {
      callback(err);
    } else {
      callback(null);
    }
  });
}

// 自定义 Readable 流类
class MyReadable extends Readable {
  constructor(filename, options) {
    super(options);
    this.filename = filename;
    this.fd = null;
    this.buffer = Buffer.alloc(1024);
    this.position = 0;
    this.open();
  }

  open() {
    console.log('Readable打开文件:', this.filename);
    checkReadPermission(this.filename, (err) => {
      if (err) {
        this.emit('error', err);
        return;
      }
      fs.open(this.filename, 'r', (err, fd) => {
        console.log('fs.open 回调:', err, fd);
        if (err) {
          this.emit('error', err);
          return;
        }
        this.fd = fd;
        this.emit('open'); // 发射 open 事件
        this.read();
      });
    });
  }

  read() {
    // console.log('读取文件:', this.position);
    if (!this.fd) return;
    fs.read(this.fd, this.buffer, 0, this.buffer.length, this.position, (err, bytesRead) => {
      if (err) {
        this.emit('error', err);
        return;
      }
      if (bytesRead === 0) {
        fs.close(this.fd, (err) => {
          if (err) this.emit('error', err);
          this.push(null); // No more data
        });
        return;
      }
      this.position += bytesRead;
      this.push(this.buffer.slice(0, bytesRead));
    });
  }

  _read(size) {
    console.log('_read');
    this.read();
  }
}

// 自定义 Writable 流类
class MyWritable extends Writable {
  constructor(filename, options) {
    super(options);
    this.filename = filename;
    this.fd = null;
    this.open();
  }

  open() {
    console.log('Writable打开文件:', this.filename);
    const dirPath = path.dirname(this.filename);
    checkWritePermission(dirPath, (err) => {
      if (err) {
        this.emit('error', err);
        return;
      }
      fs.open(this.filename, 'w', (err, fd) => {
        console.log('fs.open 回调:', err, fd);
        if (err) {
          this.emit('error', err);
          return;
        }
        this.fd = fd;
        this.emit('open'); // 发射 open 事件
      });
    });
  }

  _write(chunk, encoding, callback) {
    if (!this.fd) {
      this.once('open', () => this._write(chunk, encoding, callback));
      return;
    }
    fs.write(this.fd, chunk, 0, chunk.length, null, (err, bytesWritten) => {
      if (err) {
        this.emit('error', err);
        callback(err);
        return;
      }
      callback();
    });
  }

  _final(callback) {
    if (!this.fd) {
      this.once('open', () => this._final(callback));
      return;
    }
    fs.close(this.fd, (err) => {
      if (err) this.emit('error', err);
      callback(err);
    });
  }
}

// 创建自定义的 Readable 和 Writable 流实例
const myReadable = new MyReadable('./input.txt');
const myWritable = new MyWritable('./output.txt');

// 将 Readable 流通过管道传递给 Writable 流

// 监听错误事件
myReadable.on('error', (err) => {
  console.error('读取文件时出错:', err);
});

myWritable.on('error', (err) => {
  console.error('写入文件时出错:', err);
});

myWritable.on('finish', () => {
  console.log('文件传输完成');
});

// 添加日志以调试
myReadable.on('open', () => {
  console.log('Readable 流已打开');
});

myWritable.on('open', () => {
  console.log('Writable 流已打开');
});

myReadable.on('data', (chunk) => {
  console.log(`Received chunk: ${chunk.length} bytes`);
});

myWritable.on('drain', () => {
  console.log('Writable 流已清空');
});
